package com.sys.tdhclient.operate;

import com.sys.tdhclient.utils.LoginUtil;
import org.apache.commons.lang.StringUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.*;
import org.apache.hadoop.io.IOUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.File;
import java.io.FileInputStream;
import java.io.IOException;
import java.io.Serializable;
import java.util.Arrays;
import java.util.Properties;

public class HdfsOperate implements Serializable {

    private static final long serialVersionUID = 1L;

    private static final Logger LOG = LoggerFactory.getLogger(HdfsOperate.class);


    private static Configuration conf;
    private static String PATH_TO_HDFS_SITE_XML;
    private static String PATH_TO_CORE_SITE_XML;
    private static String PRNCIPAL_NAME;
    private static String PATH_TO_KEYTAB ;
    private static String PATH_TO_KRB5_CONF ;
    public static FileSystem fileSystem;
    public void initHdfs(String confFilePath) throws Exception{
        File confFile = new File(confFilePath +"conf.properties");
        if(confFile.exists())
        {
            Properties confProps = new Properties();
            confProps.load(new FileInputStream(confFilePath +"conf.properties"));
            PRNCIPAL_NAME = confProps.getProperty("userPrincipal");
        }
        PATH_TO_HDFS_SITE_XML = confFilePath +"hdfs-site.xml";
        PATH_TO_CORE_SITE_XML = confFilePath +"core-site.xml";
        PATH_TO_KEYTAB =confFilePath +"keytab" ;
        PATH_TO_KRB5_CONF = confFilePath +"krb5.conf";
        conf = new Configuration();
        conf.addResource(new Path(PATH_TO_HDFS_SITE_XML));
        conf.addResource(new Path(PATH_TO_CORE_SITE_XML));
        conf.set("fs.hdfs.impl", "org.apache.hadoop.hdfs.DistributedFileSystem");
        conf.set("fs.hdfs.impl.disable.cache","true");
        LOG.info("集群登录认证。。。。");
        //集群认证
        System.setProperty("HADOOP_USER_NAME",PRNCIPAL_NAME);
        if ("kerberos".equalsIgnoreCase(conf.get("hadoop.security.authentication"))) {
            System.setProperty("java.security.krb5.conf", PATH_TO_KRB5_CONF);
            try {
                LoginUtil.login(PRNCIPAL_NAME, PATH_TO_KEYTAB, PATH_TO_KRB5_CONF, conf);
            } catch (IOException e) {
                LOG.error("集群登录认证失败。。。", e);
                throw new Exception("集群登录认证失败！",e);
            }
        }
        LOG.info("登录成功...");
    }


    public  void deleteHdfsFile(String path) throws Exception {
        FileSystem fileSystem = null;
        Path hdfsPath = new Path(path);
        try {
            fileSystem = FileSystem.get(conf);
            if (fileSystem.exists(hdfsPath)) {
                LOG.info("删除hdfs文件："+path);
                fileSystem.delete(hdfsPath, true);
            }
        } catch (Exception e) {
            throw new Exception("hdfs文件执行删除失败。。。", e);
        } finally{
            if(fileSystem != null){
                fileSystem.close();
            }
        }

    }




    public  synchronized boolean uploadLocalFile2HDFS(String localFile, String hdfsFile) throws Exception {
        if (StringUtils.isBlank(localFile) || StringUtils.isBlank(hdfsFile)) {
            return false;
        }
        FileSystem fileSystem = FileSystem.get(conf);
        Path src = new Path(localFile);
        Path dst = new Path(hdfsFile);
        fileSystem.copyFromLocalFile(false, src, dst);//true为删除源文件
        LOG.info("uploadLocalFile2HDFS---localFile:{},hdfsFile:{}",localFile,hdfsFile);
        if(fileSystem != null){
            fileSystem.close();
        }
        return true;
    }


    public  synchronized boolean downloadHDFSFile2Local(String hdfsFile, String localFile, boolean deleteSrc) throws Exception{
        if(StringUtils.isBlank(localFile) || StringUtils.isBlank(hdfsFile)){
            return false;
        }
        FileSystem fileSystem = FileSystem.get(conf);
        Path src = new Path(hdfsFile);
        Path dst = new Path(localFile);
        fileSystem.copyToLocalFile(deleteSrc, src, dst);
        LOG.info("downloadHDFSFile2Local---localFile:{},hdfsFile:{},deleteSrc:{}",localFile,hdfsFile,deleteSrc);
        if(fileSystem != null){
            fileSystem.close();
        }
        return true;
    }
    @SuppressWarnings("deprecation")
    public  void mergeHdfsFileByEmpty(String srcDir, String distDir) throws Exception{
        boolean flag = false;
        Path src = new Path(srcDir);
        Path dist = new Path(distDir);
        FileSystem fileSystem = null;
        try {
            fileSystem = FileSystem.get(conf);
            Long length = fileSystem.getContentSummary(new Path(srcDir)).getLength();
            if(length == 0){
                fileSystem.delete(dist);
                fileSystem.createNewFile(dist);
            }else{
                flag = copyMerge(src.getFileSystem(conf), src, dist.getFileSystem(conf), dist, false, conf, null);
            }
        } catch (IOException e) {
            LOG.error(e.getMessage());
            throw new Exception("hdfs文件执行合并失败。。。", e);
        }finally{
            if(fileSystem != null){
                fileSystem.close();
            }
        }
    }

    public static boolean copyMerge(FileSystem srcFS, Path srcDir, FileSystem dstFS, Path dstFile, boolean deleteSource, Configuration conf, String addString) throws IOException {
        dstFile = checkDest(srcDir.getName(), dstFS, dstFile, false);
            FSDataOutputStream out = dstFS.create(dstFile);
            try {
                FileStatus[] contents = srcFS.listStatus(srcDir);
                Arrays.sort(contents);

                for(int i = 0; i < contents.length; ++i) {
                    if (contents[i].isFile()) {
                        FSDataInputStream in = srcFS.open(contents[i].getPath());

                        try {
                            IOUtils.copyBytes(in, out, conf, false);
                            if (addString != null) {
                                out.write(addString.getBytes("UTF-8"));
                            }
                        } finally {
                            in.close();
                        }
                    }
                }
            } finally {
                out.close();
            }

            return deleteSource ? srcFS.delete(srcDir, true) : true;

    }
    private static Path checkDest(String srcName, FileSystem dstFS, Path dst, boolean overwrite) throws IOException {
        if (dstFS.exists(dst)) {
            FileStatus sdst = dstFS.getFileStatus(dst);
            if (sdst.isDirectory()) {
                if (null == srcName) {
                    throw new IOException("Target " + dst + " is a directory");
                }

                return checkDest((String)null, dstFS, new Path(dst, srcName), overwrite);
            }

            if (!overwrite) {
                throw new IOException("Target " + dst + " already exists");
            }
        }

        return dst;
    }


    public static void createVoidFile(String path) throws IOException {
        File file =  new File(path);
        file.createNewFile();
    }

}
